package org.aksw.xoperator.sparql;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * provides mechanisms for parallel querying of sparql endpoints. Implementeing class can make use of the thread polling by providing FutureResults that query the respective store.
 * @author joerg
 *
 */
public abstract class AbstractThreadPooledSparqler implements ISparqlEndpoint {
	
	private static Log log = LogFactory.getLog(AbstractThreadPooledSparqler.class);
	
	ThreadPoolExecutor executor;
	
	/* (non-Javadoc)
	 * @see org.aksw.xoperator.sparql.ISparqlEndpoint#process(org.aksw.xoperator.sparql.SparqlQuery)
	 */
	public abstract void process(SparqlQuery query);
	
	public abstract List<SparqlExecutionFuture<SparqlResult>> getFuturesFor(SparqlQuery query);
	
	public AbstractThreadPooledSparqler(ThreadPoolExecutor executor) {
		super();
		this.executor = executor;
	}

	public void processFutures(List<SparqlExecutionFuture<SparqlResult>> futures,int timeout){
		log.info("Executing " + futures.size()+ " queries with an timeout of: " + timeout );
		//execute them
		for (SparqlExecutionFuture<SparqlResult> futureQuery : futures) {
			executor.execute(futureQuery);
			log.info("Querying " + futureQuery.getStore()+ " for " + futureQuery.getQuery());
		}
		
		//wait for the remote results
		long start = System.currentTimeMillis();
		for (SparqlExecutionFuture<SparqlResult> futureTask : futures) {

			if(System.currentTimeMillis()-start<timeout){
				try {
					futureTask.get(timeout+start-System.currentTimeMillis(), TimeUnit.MILLISECONDS);
					log.info("Got a result from : " +  futureTask.getStore());
				} catch (InterruptedException e) {
					log.error("Interruption Error of" + futureTask.getStore() + " :", e);
				} catch (ExecutionException e) {
					log.error("Execution Exception of" + futureTask.getStore() + " :", e);
				} catch (TimeoutException e) {
					log.error("TimeoutException of" + futureTask.getStore() + " :", e);
				}
			}
			
		}
	}
	

	
	

}
